use std::thread;

use message_io::{
    network::{Endpoint, NetEvent, Transport},
    node::{self, NodeHandler},
};
use tokio::{
    sync::{mpsc, oneshot},
    task::spawn_blocking,
};

use crate::{Error, Result};

pub async fn listen(uri: &str) -> Result<(Sender, Reciever)> {
    create(Some(uri)).await
}

pub async fn connect(uri: &str) -> Result<(Sender, Reciever, ConnectionId)> {
    let (sender, reciever) = create(None).await?;
    let conn_id = sender.connect(uri).await?;

    Ok((sender, reciever, conn_id))
}

async fn create(uri: Option<&str>) -> Result<(Sender, Reciever)> {
    let (handler, listener) = node::split::<()>();

    let (s_tx, mut s_rx) = mpsc::channel(1);
    let sender = Sender {
        tx: s_tx,
        handler: handler.clone(),
    };

    let (o_tx, o_rx) = oneshot::channel();

    let uri = uri.map(|v| v.to_string());
    let handler1 = handler.clone();
    thread::spawn(move || {
        if let Some(uri) = uri {
            if let Err(e) = handler1
                .network()
                .listen(Transport::FramedTcp, uri.as_str())
                .map_err(|e| Error::from(e))
            {
                o_tx.send(Err(e)).ok();
                return;
            }
        }

        o_tx.send(Ok(())).ok();

        loop {
            if let Some((conn_id, payload)) = s_rx.blocking_recv() {
                handler1
                    .network()
                    .send(conn_id.endpoint.clone(), &payload[..]);
            } else {
                break;
            }
        }
    });

    o_rx.await.unwrap()?;

    let (r_tx, r_rx) = mpsc::channel(1);
    let reciever = Reciever {
        rx: r_rx,
        handler: handler.clone(),
    };

    thread::spawn(move || {
        listener.for_each(move |event| match event.network() {
            NetEvent::Message(endpoint, payload) => {
                if let Err(_) = r_tx.blocking_send((ConnectionId { endpoint }, payload.to_vec())) {
                    handler.stop();
                }
            }

            NetEvent::Disconnected(endpoint) => {
                if let Err(_) = r_tx.blocking_send((ConnectionId { endpoint }, Vec::new())) {
                    handler.stop();
                }
            }

            _ => {}
        });
    });

    Ok((sender, reciever))
}

#[derive(Clone)]
pub struct Sender {
    tx: mpsc::Sender<(ConnectionId, Vec<u8>)>,
    handler: NodeHandler<()>,
}

impl Sender {
    pub async fn connect(&self, uri: &str) -> Result<ConnectionId> {
        let uri = uri.to_string();
        let handler = self.handler.clone();
        spawn_blocking(move || -> Result<ConnectionId> {
            handler
                .network()
                .connect_sync(Transport::FramedTcp, uri.as_str())
                .map(|(endpoint, _)| ConnectionId { endpoint })
                .map_err(|e| Error::from(e))
        })
        .await
        .unwrap()
    }

    pub async fn send(&self, conn_id: ConnectionId, payload: Vec<u8>) -> Result<()> {
        self.tx
            .send((conn_id, payload))
            .await
            .map_err(|e| Error::from(e))
    }
}

pub struct Reciever {
    rx: mpsc::Receiver<(ConnectionId, Vec<u8>)>,
    handler: NodeHandler<()>,
}

impl Reciever {
    pub async fn recv(&mut self) -> Result<(ConnectionId, Vec<u8>)> {
        self.rx
            .recv()
            .await
            .map_or(Err(Error::new("Finished".into())), |r| Ok(r))
    }
}

impl Drop for Reciever {
    fn drop(&mut self) {
        self.handler.stop();
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct ConnectionId {
    endpoint: Endpoint,
}

#[cfg(test)]
mod test {
    use super::*;

    #[tokio::test]
    async fn test() -> Result<()> {
        let uri = "127.0.0.1:9000";
        let (s_tx, mut s_rx) = listen(uri).await?;
        let (c_tx, mut c_rx, s_conn_id) = connect(uri).await?;
        println!("s_conn_id: {:?}", s_conn_id);

        c_tx.send(s_conn_id, [0, 1, 2].to_vec()).await?;
        let (c_conn_id, payload) = s_rx.recv().await?;
        println!("c_conn_id: {:?}", c_conn_id);
        println!("payload: {:?}", payload);

        s_tx.send(c_conn_id, [3, 4, 5].to_vec()).await?;
        let (s_conn_id, payload) = c_rx.recv().await?;
        println!("s_conn_id: {:?}", s_conn_id);
        println!("payload: {:?}", payload);

        Ok(())
    }

    #[tokio::test]
    async fn test_performance_server() -> Result<()> {
        let (s_tx, mut s_rx) = listen("0.0.0.0:9000").await?;
        let (c_conn_id, _) = s_rx.recv().await?;

        let mut w = 0usize;
        let mut wi = 0usize;
        let mut t = std::time::Instant::now();
        loop {
            let buf = vec![0u8; 128 * 1024];
            s_tx.send(c_conn_id.clone(), buf.clone()).await?;

            w += buf.len();
            let i = w / (1024 * 1024);
            if i > wi + 10 {
                let cur_t = std::time::Instant::now();
                println!(
                    "speed: {}MB/s",
                    ((i - wi) as f64) / (cur_t - t).as_secs_f64()
                );
                wi = i;
                t = cur_t;
            }
        }
    }

    #[tokio::test]
    async fn test_performance_client() -> Result<()> {
        let (c_tx, mut c_rx, s_conn_id) = connect("172.16.34.76:9000").await?;
        c_tx.send(s_conn_id.clone(), [0].to_vec()).await?;

        loop {
            c_rx.recv().await?;
        }
    }
}
